package flinksql;

import lombok.Data;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

public class FlinkSqlDemo {

    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        EnvironmentSettings settings = EnvironmentSettings.newInstance()
//                .useBlinkPlanner()
//                .inStreamingMode()
//                .build();
        env.setParallelism(1);
//        env.enableCheckpointing(10000);
//        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env,settings);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //cdc
        tableEnv.executeSql("create table ods_order_info ("+
                " id int,"+
                " consignee string," +
                " consignee_tel string," +
                " total_amount decimal(10,2), " +
                " order_status string, " +
                " user_id int," +
                " payment_way string," +
                " delivery_address string," +
                " order_comment string," +
                " out_trade_no string," +
                " trade_body string," +
                " create_time string," +
                " operate_time string," +
                " expire_time string," +
                " process_status string," +
                " tracking_no string," +
                " parent_order_id int," +
                " img_url string," +
                " province_id int," +
                " activity_reduce_amount decimal(16,2)," +
                " coupon_reduce_amount decimal(16,2)," +
                " original_total_amount decimal(16,2)," +
                " feight_fee decimal(16,2)," +
                " feight_fee_reduce decimal(16,2)," +
                " refundable_time string," +
                " primary key(id) not enforced" +
                ") with (" +
                " 'connector'='mysql-cdc'," +
                " 'hostname'='192.168.45.132'," +
                " 'port'='3306'," +
                " 'username'='root'," +
                " 'password'='123456'," +
                " 'database-name'='gmall_flink'," +
                " 'table-name'='order_info'" +
                ")");


//        tableEnv.executeSql("create table base_region (" +
//                " id string,"+
//                " region_name string," +
//                " primary key (id) not ENFORCED" +
//                ") with (" +
//                " 'connector'='jdbc'," +
//                " 'url'='jdbc:mysql://192.168.45.132:3306/gmall_flink?user=root&password=123456&serverTimezone=UTC'," +
//                " 'table-name'='base_region'," +
//                " 'lookup.cache.max-rows'='5000'," +
//                " 'lookup.cache.ttl'='10s'" +
//                ")");

        Table resultTable = tableEnv.sqlQuery("select id,order_status from ods_order_info");
        tableEnv.toRetractStream(resultTable,Row.class).print();

        env.execute();

    }


    @Data
    public static class StatusCount {
        public String order_status;
        public long cnt;
    }


}
